package com.ck.framework.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;


/**
 * @author ck
 */
@Configuration
public class RabbitMqConfig {

    @Value("${spring.rabbitmq.addresses}")
    private String address;

    @Value("${spring.rabbitmq.username}")
    private String userName;

    @Value("${spring.rabbitmq.password}")
    private String passwd;

    @Value("${spring.rabbitmq.virtual-host}")
    private String portalVirHost;

    @Value("${spring.rabbitmq.biz-vir-host}")
    private String bizVirHost;


    /**
     * 简单模式
     */
    @Bean
    public Queue helloQueue() {
        return new Queue("boot-hello-queue");
    }

    /**
     * 工作模式
     */
    @Bean
    public Queue workQueue() {
        return new Queue("boot-work-queue");
    }



    /**
     * Fanout广播模式
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("boot-log-fanout-exchange");
    }

    @Bean
    public Queue fanoutQueue1() {
        return new Queue("boot-log-fanout-queue1");
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue("boot-log-fanout-queue2");
    }

    /**
     * 将log-fanout-queue1、log-fanout-queue2两个队列绑定到boot-log-fanout-exchange交换机上面，发送端的routing_key写任何字符都会被忽略：
     */
    @Bean
    public Binding bindingFanoutExchangeAndFanoutQueue1(FanoutExchange fanoutExchange, Queue fanoutQueue1) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    @Bean
    public Binding bindingFanoutExchangeAndFanoutQueue2(FanoutExchange fanoutExchange, Queue fanoutQueue2) {
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }



    /**
     * Direct路由模式
     */
    @Bean
    public DirectExchange directexchange() {
        return new DirectExchange("boot-log-direct-exchange");
    }

    @Bean
    public Queue directQueue1() {
        return new Queue("boot-log-direct-queue-info");
    }

    @Bean
    public Queue directQueue2() {
        return new Queue("boot-log-direct-queue-warn");
    }

    @Bean
    public Queue directQueue3() {
        return new Queue("boot-log-direct-queue-error");
    }

    @Bean
    public Binding bindingDirectExchangeAndDirectQueue1(DirectExchange directexchange, Queue directQueue1) {
        return BindingBuilder.bind(directQueue1).to(directexchange).with("info");
    }

    @Bean
    public Binding bindingDirectExchangeAndDirectQueue2(DirectExchange directexchange, Queue directQueue2) {
        return BindingBuilder.bind(directQueue2).to(directexchange).with("warning");
    }

    @Bean
    public Binding bindingDirectExchangeAndDirectQueue3(DirectExchange directexchange, Queue directQueue3) {
        return BindingBuilder.bind(directQueue3).to(directexchange).with("error");
    }



    /**
     * Topic 主题模式（*表示一个词  #表示零个或多个词）
     */
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("boot-log-topic-exchange");
    }

    @Bean
    public Queue topicQueue1() {
        return new Queue("boot-log-topic-queue1");
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue("boot-log-topic-queue2");
    }

    @Bean
    public Binding bindingTopicExchangeAndTopicQueue1(TopicExchange topicExchange, Queue topicQueue1) {
        return BindingBuilder.bind(topicQueue1).to(topicExchange).with("*.critical");
    }

    @Bean
    public Binding bindingTopicExchangeAndTopicQueue2(TopicExchange topicExchange, Queue topicQueue2) {
        return BindingBuilder.bind(topicQueue2).to(topicExchange).with("kernel.*");
    }

    /**
     * 测试消费者消息确认模式为AUTO模式，开启消息重试，重试次数用完后的失败策略：RepublishMessageRecoverer
     * 1、定义接收失败消息的交换机、队列及其绑定关系
     * 2、定义 RepublishMessageRecoverer
     * 3、失败后将消息投递到一个指定的，专门存放异常消息的队列(RepublishMessageRecoverer)，后续由人工集中处理
     */
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }

    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue");
    }

    @Bean
    public Binding errorMessageBinding(DirectExchange errorMessageExchange, Queue errorQueue ){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

    /**
     * 消费重试拒绝策略
     * @param rabbitTemplate
     * @return
     */
    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }




    /**
     * TTL + 死信队列 实现延时消息-【声明一组普通交换机和队列】
     */
    @Bean
    public DirectExchange ttlDirectExchange(){
        return new DirectExchange("ttl.direct");
    }

    @Bean
    public Queue ttlQueue(){
        return QueueBuilder
                // 指定队列名称，并持久化
                .durable("ttl.queue")
                //设置队列的超时时间，10秒
                .ttl(10000)
                // 指定死信交换机
                .deadLetterExchange("dl.direct")
                //指定死信routingKey
                .deadLetterRoutingKey("dl")
                .build();
    }

    @Bean
    public Binding ttlBinding(){
        return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");
    }

    /**
     * TTL + 死信队列 实现延时消息-【声明一组死信交换机和队列】
     */
    @Bean
    public DirectExchange dlMessageExchange(){
        return new DirectExchange("dl.direct");
    }

    @Bean
    public Queue dlQueue(){
        return new Queue("dl.queue");
    }

    @Bean
    public Binding dlMessageBinding(){
        return BindingBuilder.bind(dlQueue()).to(dlMessageExchange()).with("dl");
    }


    /**
     * DelayExchange 延时插件实现 延时消息
     */
    @Bean
    public DirectExchange delayedExchange(){
        return ExchangeBuilder.directExchange("delay.direct").delayed().build();
    }

    @Bean
    public Queue delayQueue(){
        return new Queue("delay.queue");
    }

    @Bean
    public Binding delayedBinding(){
        return BindingBuilder.bind(delayQueue()).to(delayedExchange()).with("delay");
    }

    /**
     * 声明惰性队列（两种方式）
     * 1、基于@Bean声明lazy-queue
     * 2、基于@RabbitListener声明LazyQueue
     */
    @Bean
    public Queue lazyQueue() {
        return QueueBuilder.durable("lazy.queue")
                .lazy()  //开启x-queue-mode为lazy
                .build();
    }

    @Bean
    public Queue normalQueue() {
        return QueueBuilder.durable("normal.queue")
                .build();
    }



    /**
     * 主链接
     *
     * @return
     */
    @Bean(name = "portalConnFac")
    @Primary
    public ConnectionFactory portalConnFac() {
        CachingConnectionFactory connFac = new CachingConnectionFactory();
        connFac.setAddresses(address);
        connFac.setUsername(userName);
        connFac.setPassword(passwd);
        connFac.setVirtualHost(portalVirHost);
        return connFac;
    }

    /**
     * 副链接
     *
     * @return
     */
    @Bean(name = "bizConnFac")
    public ConnectionFactory bizConnFac() {
        CachingConnectionFactory connFac = new CachingConnectionFactory();
        connFac.setAddresses(address);
        connFac.setUsername(userName);
        connFac.setPassword(passwd);
        connFac.setVirtualHost(bizVirHost);
        return connFac;
    }

    /**
     * 用Jackson2JsonMessageConverter替换RabbitMq基于JDK的默认的消息转换器：SimpleMessageConverter（可读性差、传输效率低）
     * @param connFac
     * @return
     */
    @Bean
    @Primary
    public RabbitTemplate portalTemp(@Qualifier("portalConnFac") ConnectionFactory connFac) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connFac);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());  // 使用 Jackson2JsonMessage 转换器
        return rabbitTemplate;
    }

    @Bean(name = "bizTemp")
    public RabbitTemplate bizTemp(@Qualifier("bizConnFac") ConnectionFactory connFac) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connFac);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());  // 使用 Jackson2JsonMessage 转换器
        return rabbitTemplate;
    }

    /**
     * 批量消费
     *
     * @param configurer
     * @param connFac
     * @return
     */
    @Bean("batchFac")
    public RabbitListenerContainerFactory batchFac(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("portalConnFac") ConnectionFactory connFac) {

        // 自动装配属性
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connFac);

        // 批量消费者模式必须开启, 否则抛出异常
        factory.setBatchListener(true);                                     //配置监听器容器工厂为批量消息监听器模式。
        factory.setConsumerBatchEnabled(true);                              //启用批量消费者模式
        factory.setDeBatchingEnabled(true);                                 //启用消息拆分
        factory.setMessageConverter(new Jackson2JsonMessageConverter());       //设置消息转换器
        return factory;
    }


    /**
     * 单条消费
     *
     * @param configurer
     * @param connFac
     * @return
     */
    @Bean("singleFac")
    public RabbitListenerContainerFactory singleFac(SimpleRabbitListenerContainerFactoryConfigurer configurer,  @Qualifier("portalConnFac") ConnectionFactory connFac) {

        // 自动装配属性
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connFac);

        factory.setBatchListener(false);
        factory.setConsumerBatchEnabled(false);
        factory.setDeBatchingEnabled(false);
        factory.setBatchSize(1);

        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        return factory;
    }



}
